package com.lock.demo.zookeeper.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * @ProjectName: lock_demo
 * @Package: com.lock.demo.zookeeper.lock
 * @Author: huat
 * @Date: 2021/5/5 14:56
 * @Version: 1.0
 */
public class ZKLock implements Lock {
    //zookeeper客户端
    private  ZooKeeper zk;
    //zk目录结构
    private final static String root="/locks";
    //锁名称
    private String lockName;
    //当前线程创建序列化节点
    private ThreadLocal<String> threadLocal=new ThreadLocal<String>();
    //用来同步等待zkclient连接到服务器
    private CountDownLatch countDownLatch=new CountDownLatch(1);
    //会话超时时间
    private final static int sessionTimeOut=3000;
    private final static byte[] data=new byte[0];

    public ZKLock(String config,String lockName) {
        this.lockName = lockName;
        try{
            zk=new ZooKeeper(config, sessionTimeOut, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    //建立连接
                    if(watchedEvent.getState()== Event.KeeperState.SyncConnected){
                        countDownLatch.countDown();
                    }
                }
            });
            countDownLatch.await();
            Stat exists = zk.exists(root, false);
            if(null==exists){
                //创建根节点
                zk.create(root,data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }

        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void lock() {
        try {
            //创建临时节点
            String myNode = zk.create(root + "/" + lockName, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            //获取所有子节点
            List<String> children = zk.getChildren(root, false);
            TreeSet<String> nodes=new TreeSet<String>();
            //排序子节点
            for (String str:children) {
                nodes.add(root+"/"+str);
            }
            String smallNode=nodes.first();
            if(myNode.equals(smallNode)){
                //如果最小的节点则表示获取锁，获取锁 后跳出
                this.threadLocal.set(myNode);

            }else{
                //如果不是最小节点则获取前一个节点
                String preNode=nodes.lower(myNode);

                CountDownLatch countDownLatch1=new CountDownLatch(1);
                //同时注册监听
                Stat exists = zk.exists(preNode, new LockWatch(countDownLatch1));
                //判断比自己小一个数的节点是否存在，如果不存在则无须等待直接获取锁，同时获取监听
                if(null!=exists){
                    countDownLatch1.await();//等待其他线程释放锁
                    threadLocal.set(myNode);
                    countDownLatch=null;
                }
            }

        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock() {
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        try {
            zk.delete(threadLocal.get(),-1);
            threadLocal.remove();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }
}
